/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.apache.iotdb.confignode.service.thrift;

import org.apache.iotdb.common.rpc.thrift.*;
import org.apache.iotdb.commons.auth.AuthException;
import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.commons.consensus.ConsensusGroupId;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.path.PathPatternTree;
import org.apache.iotdb.commons.utils.StatusUtils;
import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.confignode.conf.ConfigNodeConstant;
import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
import org.apache.iotdb.confignode.conf.SystemPropertiesUtils;
import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlanType;
import org.apache.iotdb.confignode.consensus.request.auth.AuthorPlan;
import org.apache.iotdb.confignode.consensus.request.read.datanode.GetDataNodeConfigurationPlan;
import org.apache.iotdb.confignode.consensus.request.read.partition.GetDataPartitionPlan;
import org.apache.iotdb.confignode.consensus.request.read.partition.GetOrCreateDataPartitionPlan;
import org.apache.iotdb.confignode.consensus.request.read.partition.GetSeriesSlotListPlan;
import org.apache.iotdb.confignode.consensus.request.read.partition.GetTimeSlotListPlan;
import org.apache.iotdb.confignode.consensus.request.read.region.GetRegionInfoListPlan;
import org.apache.iotdb.confignode.consensus.request.read.storagegroup.CountStorageGroupPlan;
import org.apache.iotdb.confignode.consensus.request.read.storagegroup.GetStorageGroupPlan;
import org.apache.iotdb.confignode.consensus.request.write.confignode.RemoveConfigNodePlan;
import org.apache.iotdb.confignode.consensus.request.write.datanode.RegisterDataNodePlan;
import org.apache.iotdb.confignode.consensus.request.write.datanode.RemoveDataNodePlan;
import org.apache.iotdb.confignode.consensus.request.write.datanode.UpdateDataNodePlan;
import org.apache.iotdb.confignode.consensus.request.write.storagegroup.*;
import org.apache.iotdb.confignode.consensus.request.write.sync.CreatePipeSinkPlan;
import org.apache.iotdb.confignode.consensus.request.write.sync.DropPipeSinkPlan;
import org.apache.iotdb.confignode.consensus.response.*;
import org.apache.iotdb.confignode.manager.ConfigManager;
import org.apache.iotdb.confignode.manager.ConsensusManager;
import org.apache.iotdb.confignode.rpc.thrift.*;
import org.apache.iotdb.confignode.service.ConfigNode;
import org.apache.iotdb.consensus.common.response.ConsensusGenericResponse;
import org.apache.iotdb.db.mpp.plan.statement.AuthorType;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.thrift.TException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Collections;
import java.util.List;

/**
 * ConfigNodeRPCServer exposes the interface that interacts with the DataNode
 */
public class ConfigNodeRPCServiceProcessor implements IConfigNodeRPCService.Iface {

    private static final Logger LOGGER = LoggerFactory.getLogger(ConfigNodeRPCServiceProcessor.class);

    private final ConfigManager configManager;

    public ConfigNodeRPCServiceProcessor(ConfigManager configManager) {
        this.configManager = configManager;
    }

    @TestOnly
    public void close() throws IOException {
        configManager.close();
    }

    @TestOnly
    public ConsensusManager getConsensusManager() {
        return configManager.getConsensusManager();
    }

    @Override
    public TSystemConfigurationResp getSystemConfiguration() {
        TSystemConfigurationResp resp =
                ((ConfigurationResp) configManager.getSystemConfiguration())
                        .convertToRpcSystemConfigurationResp();

        // Print log to record the ConfigNode that performs the GetConfigurationRequest
        LOGGER.info("Execute GetSystemConfiguration with result {}", resp);
        return resp;
    }

    @Override
    public TDataNodeRegisterResp registerDataNode(TDataNodeRegisterReq req) {
        TDataNodeRegisterResp resp =
                ((DataNodeRegisterResp)
                        configManager.registerDataNode(
                                new RegisterDataNodePlan(req.getDataNodeConfiguration())))
                        .convertToRpcDataNodeRegisterResp();

        // Print log to record the ConfigNode that performs the RegisterDatanodeRequest
        LOGGER.info("Execute RegisterDatanodeRequest {} with result {}", req, resp);

        return resp;
    }

    @Override
    public TDataNodeRestartResp restartDataNode(TDataNodeRestartReq req) {
        TDataNodeRestartResp resp = configManager.restartDataNode(req);

        // Print log to record the ConfigNode that performs the RestartDatanodeRequest
        LOGGER.info("Execute RestartDatanodeRequest {} with result {}", req, resp);

        return resp;
    }

    @Override
    public TDataNodeRemoveResp removeDataNode(TDataNodeRemoveReq req) {
        LOGGER.info("ConfigNode RPC Service start to remove DataNode, req: {}", req);
        RemoveDataNodePlan removeDataNodePlan = new RemoveDataNodePlan(req.getDataNodeLocations());
        DataNodeToStatusResp removeResp =
                (DataNodeToStatusResp) configManager.removeDataNode(removeDataNodePlan);
        TDataNodeRemoveResp resp = removeResp.convertToRpCDataNodeRemoveResp();
        LOGGER.info(
                "ConfigNode RPC Service finished to remove DataNode, req: {}, result: {}", req, resp);
        return resp;
    }

    @Override
    public TDataNodeRegisterResp updateDataNode(TDataNodeUpdateReq req) {
        LOGGER.info("ConfigNode RPC Service start to update DataNode, req: {}", req);
        UpdateDataNodePlan updateDataNodePlan = new UpdateDataNodePlan(req.getDataNodeLocation());
        TDataNodeRegisterResp resp =
                ((DataNodeRegisterResp) configManager.updateDataNode(updateDataNodePlan))
                        .convertToRpcDataNodeRegisterResp();
        LOGGER.info(
                "ConfigNode RPC Service finished to update DataNode, req: {}, result: {}", req, resp);
        return resp;
    }

    @Override
    public TDataNodeConfigurationResp getDataNodeConfiguration(int dataNodeID) {
        GetDataNodeConfigurationPlan queryReq = new GetDataNodeConfigurationPlan(dataNodeID);
        DataNodeConfigurationResp queryResp =
                (DataNodeConfigurationResp) configManager.getDataNodeConfiguration(queryReq);

        TDataNodeConfigurationResp resp = new TDataNodeConfigurationResp();
        queryResp.convertToRpcDataNodeLocationResp(resp);
        return resp;
    }

    @Override
    public TSStatus reportRegionMigrateResult(TRegionMigrateResultReportReq req) {
        return configManager.reportRegionMigrateResult(req);
    }

    @Override
    public TShowClusterResp showCluster() {
        return configManager.showCluster();
    }

    @Override
    public TShowClusterParametersResp showClusterParameters() throws TException {
        return configManager.showClusterParameters();
    }

    @Override
    public TSStatus setStorageGroup(TSetStorageGroupReq req) throws TException {
        TStorageGroupSchema storageGroupSchema = req.getStorageGroup();

        // Set default configurations if necessary
        if (!storageGroupSchema.isSetTTL()) {
            storageGroupSchema.setTTL(CommonDescriptor.getInstance().getConfig().getDefaultTTLInMs());
        }
        if (!storageGroupSchema.isSetSchemaReplicationFactor()) {
            storageGroupSchema.setSchemaReplicationFactor(
                    ConfigNodeDescriptor.getInstance().getConf().getSchemaReplicationFactor());
        }
        if (!storageGroupSchema.isSetDataReplicationFactor()) {
            storageGroupSchema.setDataReplicationFactor(
                    ConfigNodeDescriptor.getInstance().getConf().getDataReplicationFactor());
        }
        if (!storageGroupSchema.isSetTimePartitionInterval()) {
            storageGroupSchema.setTimePartitionInterval(
                    ConfigNodeDescriptor.getInstance().getConf().getTimePartitionInterval());
        }

        // Initialize the maxSchemaRegionGroupCount and maxDataRegionGroupCount as 0
        storageGroupSchema.setMaxSchemaRegionGroupNum(0);
        storageGroupSchema.setMaxDataRegionGroupNum(0);

        SetStorageGroupPlan setReq = new SetStorageGroupPlan(storageGroupSchema);
        TSStatus resp = configManager.setStorageGroup(setReq);

        // Print log to record the ConfigNode that performs the set SetStorageGroupRequest
        LOGGER.info("Execute SetStorageGroupRequest {} with result {}", req, resp);

        return resp;
    }

    @Override
    public TSStatus deleteStorageGroup(TDeleteStorageGroupReq tDeleteReq) {
        String prefixPath = tDeleteReq.getPrefixPath();
        return configManager.deleteStorageGroups(Collections.singletonList(prefixPath));
    }

    @Override
    public TSStatus deleteStorageGroups(TDeleteStorageGroupsReq tDeleteReq) {
        List<String> prefixList = tDeleteReq.getPrefixPathList();
        return configManager.deleteStorageGroups(prefixList);
    }

    @Override
    public TSStatus setTTL(TSetTTLReq req) throws TException {
        return configManager.setTTL(new SetTTLPlan(req.getStorageGroupPathPattern(), req.getTTL()));
    }

    @Override
    public TSStatus setSchemaReplicationFactor(TSetSchemaReplicationFactorReq req) throws TException {
        return configManager.setSchemaReplicationFactor(
                new SetSchemaReplicationFactorPlan(
                        req.getStorageGroup(), req.getSchemaReplicationFactor()));
    }

    @Override
    public TSStatus setDataReplicationFactor(TSetDataReplicationFactorReq req) throws TException {
        return configManager.setDataReplicationFactor(
                new SetDataReplicationFactorPlan(req.getStorageGroup(), req.getDataReplicationFactor()));
    }

    @Override
    public TSStatus setTimePartitionInterval(TSetTimePartitionIntervalReq req) throws TException {
        return configManager.setTimePartitionInterval(
                new SetTimePartitionIntervalPlan(req.getStorageGroup(), req.getTimePartitionInterval()));
    }

    @Override
    public TCountStorageGroupResp countMatchedStorageGroups(List<String> storageGroupPathPattern) {
        CountStorageGroupResp countStorageGroupResp =
                (CountStorageGroupResp)
                        configManager.countMatchedStorageGroups(
                                new CountStorageGroupPlan(storageGroupPathPattern));

        TCountStorageGroupResp resp = new TCountStorageGroupResp();
        countStorageGroupResp.convertToRPCCountStorageGroupResp(resp);
        return resp;
    }

    @Override
    public TStorageGroupSchemaResp getMatchedStorageGroupSchemas(
            List<String> storageGroupPathPattern) {
        StorageGroupSchemaResp storageGroupSchemaResp =
                (StorageGroupSchemaResp)
                        configManager.getMatchedStorageGroupSchemas(
                                new GetStorageGroupPlan(storageGroupPathPattern));

        return storageGroupSchemaResp.convertToRPCStorageGroupSchemaResp();
    }

    @Override
    public TSchemaPartitionTableResp getSchemaPartitionTable(TSchemaPartitionReq req) {
        PathPatternTree patternTree =
                PathPatternTree.deserialize(ByteBuffer.wrap(req.getPathPatternTree()));
        return configManager.getSchemaPartition(patternTree);
    }

    @Override
    public TSchemaPartitionTableResp getOrCreateSchemaPartitionTable(TSchemaPartitionReq req) {
        PathPatternTree patternTree =
                PathPatternTree.deserialize(ByteBuffer.wrap(req.getPathPatternTree()));
        return configManager.getOrCreateSchemaPartition(patternTree);
    }

    @Override
    public TSchemaNodeManagementResp getSchemaNodeManagementPartition(TSchemaNodeManagementReq req) {
        PathPatternTree patternTree =
                PathPatternTree.deserialize(ByteBuffer.wrap(req.getPathPatternTree()));
        PartialPath partialPath = patternTree.getAllPathPatterns().get(0);
        return configManager.getNodePathsPartition(partialPath, req.getLevel());
    }

    @Override
    public TDataPartitionTableResp getDataPartitionTable(TDataPartitionReq req) {
        GetDataPartitionPlan getDataPartitionPlan =
                GetDataPartitionPlan.convertFromRpcTDataPartitionReq(req);
        return configManager.getDataPartition(getDataPartitionPlan);
    }

    @Override
    public TDataPartitionTableResp getOrCreateDataPartitionTable(TDataPartitionReq req) {
        GetOrCreateDataPartitionPlan getOrCreateDataPartitionReq =
                GetOrCreateDataPartitionPlan.convertFromRpcTDataPartitionReq(req);
        return configManager.getOrCreateDataPartition(getOrCreateDataPartitionReq);
    }

    @Override
    public TSStatus operatePermission(TAuthorizerReq req) {
        if (req.getAuthorType() < 0 || req.getAuthorType() >= AuthorType.values().length) {
            throw new IndexOutOfBoundsException("Invalid Author Type ordinal");
        }
        AuthorPlan plan = null;
        try {
            plan =
                    new AuthorPlan(
                            ConfigPhysicalPlanType.values()[
                                    req.getAuthorType() + ConfigPhysicalPlanType.Author.ordinal() + 1],
                            req.getUserName(),
                            req.getRoleName(),
                            req.getPassword(),
                            req.getNewPassword(),
                            req.getPermissions(),
                            req.getNodeNameList());
        } catch (AuthException e) {
            LOGGER.error(e.getMessage());
        }
        return configManager.operatePermission(plan);
    }

    @Override
    public TAuthorizerResp queryPermission(TAuthorizerReq req) {
        if (req.getAuthorType() < 0 || req.getAuthorType() >= AuthorType.values().length) {
            throw new IndexOutOfBoundsException("Invalid Author Type ordinal");
        }
        AuthorPlan plan = null;
        try {
            plan =
                    new AuthorPlan(
                            ConfigPhysicalPlanType.values()[
                                    req.getAuthorType() + ConfigPhysicalPlanType.Author.ordinal() + 1],
                            req.getUserName(),
                            req.getRoleName(),
                            req.getPassword(),
                            req.getNewPassword(),
                            req.getPermissions(),
                            req.getNodeNameList());
        } catch (AuthException e) {
            LOGGER.error(e.getMessage());
        }
        PermissionInfoResp dataSet = (PermissionInfoResp) configManager.queryPermission(plan);
        TAuthorizerResp resp = new TAuthorizerResp(dataSet.getStatus());
        resp.setAuthorizerInfo(dataSet.getPermissionInfo());
        return resp;
    }

    @Override
    public TPermissionInfoResp login(TLoginReq req) {
        return configManager.login(req.getUserrname(), req.getPassword());
    }

    @Override
    public TPermissionInfoResp checkUserPrivileges(TCheckUserPrivilegesReq req) {
        return configManager.checkUserPrivileges(
                req.getUsername(), req.getPaths(), req.getPermission());
    }

    @Override
    public TConfigNodeRegisterResp registerConfigNode(TConfigNodeRegisterReq req) {
        TConfigNodeRegisterResp resp = configManager.registerConfigNode(req);

        // Print log to record the ConfigNode that performs the RegisterConfigNodeRequest
        LOGGER.info("Execute RegisterConfigNodeRequest {} with result {}", req, resp);

        return resp;
    }

    @Override
    public TSStatus addConsensusGroup(TAddConsensusGroupReq registerResp) {
        return configManager.createPeerForConsensusGroup(registerResp.getConfigNodeList());
    }

    @Override
    public TSStatus notifyRegisterSuccess() {
        try {
            SystemPropertiesUtils.storeSystemParameters();
        } catch (IOException e) {
            LOGGER.error("Write confignode-system.properties failed", e);
            return new TSStatus(TSStatusCode.WRITE_PROCESS_ERROR.getStatusCode());
        }

        // The initial startup of Non-Seed-ConfigNode finished
        LOGGER.info(
                "{} has successfully started and joined the cluster.", ConfigNodeConstant.GLOBAL_NAME);
        return StatusUtils.OK;
    }

    @Override
    public TSStatus restartConfigNode(TConfigNodeRestartReq req) {
        TSStatus status = configManager.restartConfigNode(req);

        // Print log to record the ConfigNode that performs the RegisterConfigNodeRequest
        LOGGER.info("Execute RestartConfigNodeRequest {} with result {}", req, status);

        return status;
    }

    /**
     * For leader to remove ConfigNode configuration in consensus layer
     */
    @Override
    public TSStatus removeConfigNode(TConfigNodeLocation configNodeLocation) throws TException {
        RemoveConfigNodePlan removeConfigNodePlan = new RemoveConfigNodePlan(configNodeLocation);
        TSStatus status = configManager.removeConfigNode(removeConfigNodePlan);
        // Print log to record the ConfigNode that performs the RemoveConfigNodeRequest
        LOGGER.info("Execute RemoveConfigNodeRequest {} with result {}", configNodeLocation, status);

        return status;
    }

    @Override
    public TSStatus deleteConfigNodePeer(TConfigNodeLocation configNodeLocation) {
        if (!configManager.getNodeManager().getRegisteredConfigNodes().contains(configNodeLocation)) {
            return new TSStatus(TSStatusCode.REMOVE_CONFIGNODE_ERROR.getStatusCode())
                    .setMessage(
                            "remove ConsensusGroup failed because the ConfigNode not in current Cluster.");
        }

        ConsensusGroupId groupId = configManager.getConsensusManager().getConsensusGroupId();
        ConsensusGenericResponse resp =
                configManager.getConsensusManager().getConsensusImpl().deletePeer(groupId);
        if (!resp.isSuccess()) {
            return new TSStatus(TSStatusCode.REMOVE_CONFIGNODE_ERROR.getStatusCode())
                    .setMessage(
                            "remove ConsensusGroup failed because internal failure. See other logs for more details");
        }

        return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())
                .setMessage("remove ConsensusGroup success.");
    }

    /**
     * stop config node
     */
    @Override
    public TSStatus stopConfigNode(TConfigNodeLocation configNodeLocation) {
        new Thread(
                () -> {
                    try {
                        ConfigNode.getInstance().stop();
                    } catch (IOException e) {
                        LOGGER.error("Meet error when stop ConfigNode!", e);
                    } finally {
                        System.exit(0);
                    }
                })
                .start();
        return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())
                .setMessage("Stop ConfigNode success.");
    }

    @Override
    public TSStatus createFunction(TCreateFunctionReq req) {
        return configManager.createFunction(req);
    }

    @Override
    public TSStatus dropFunction(TDropFunctionReq req) {
        return configManager.dropFunction(req.getUdfName());
    }

    @Override
    public TGetUDFTableResp getUDFTable() {
        return configManager.getUDFTable();
    }

    @Override
    public TGetJarInListResp getUDFJar(TGetJarInListReq req) {
        return configManager.getUDFJar(req);
    }

    @Override
    public TSStatus createTrigger(TCreateTriggerReq req) {
        return configManager.createTrigger(req);
    }

    @Override
    public TSStatus dropTrigger(TDropTriggerReq req) {
        return configManager.dropTrigger(req);
    }

    @Override
    public TGetTriggerTableResp getTriggerTable() {
        return configManager.getTriggerTable();
    }

    @Override
    public TGetTriggerTableResp getStatefulTriggerTable() {
        return configManager.getStatefulTriggerTable();
    }

    @Override
    public TGetLocationForTriggerResp getLocationOfStatefulTrigger(String triggerName) {
        return configManager.getLocationOfStatefulTrigger(triggerName);
    }

    @Override
    public TGetJarInListResp getTriggerJar(TGetJarInListReq req) {
        return configManager.getTriggerJar(req);
    }

    @Override
    public TSStatus merge() throws TException {
        return configManager.merge();
    }

    @Override
    public TSStatus flush(TFlushReq req) throws TException {
        if (req.storageGroups != null) {
            List<PartialPath> noExistSg =
                    configManager.checkStorageGroupExist(PartialPath.fromStringList(req.storageGroups));
            if (!noExistSg.isEmpty()) {
                StringBuilder sb = new StringBuilder();
                noExistSg.forEach(storageGroup -> sb.append(storageGroup.getFullPath()).append(","));
                return RpcUtils.getStatus(
                        TSStatusCode.DATABASE_NOT_EXIST,
                        "storageGroup " + sb.subSequence(0, sb.length() - 1) + " does not exist");
            }
        }
        return configManager.flush(req);
    }

    @Override
    public TSStatus clearCache() {
        return configManager.clearCache();
    }

    @Override
    public TSStatus loadConfiguration() {
        return configManager.loadConfiguration();
    }

    @Override
    public TSStatus setSystemStatus(String status) {
        return configManager.setSystemStatus(status);
    }

    @TestOnly
    @Override
    public TSStatus setDataNodeStatus(TSetDataNodeStatusReq req) {
        return configManager.setDataNodeStatus(req);
    }

    @Override
    public TGetDataNodeLocationsResp getRunningDataNodeLocations() {
        return configManager.getRunningDataNodeLocations();
    }

    @Override
    public TShowRegionResp showRegion(TShowRegionReq showRegionReq) {
        GetRegionInfoListPlan getRegionInfoListPlan = new GetRegionInfoListPlan(showRegionReq);
        RegionInfoListResp dataSet = configManager.showRegion(getRegionInfoListPlan);
        TShowRegionResp showRegionResp = new TShowRegionResp();
        showRegionResp.setStatus(dataSet.getStatus());
        showRegionResp.setRegionInfoList(dataSet.getRegionInfoList());
        return showRegionResp;
    }

    @Override
    public TRegionRouteMapResp getLatestRegionRouteMap() {
        return configManager.getLatestRegionRouteMap();
    }

    @Override
    public long getConfigNodeHeartBeat(long timestamp) {
        return timestamp;
    }

    @Override
    public TShowDataNodesResp showDataNodes() {
        return configManager.showDataNodes();
    }

    @Override
    public TShowConfigNodesResp showConfigNodes() {
        return configManager.showConfigNodes();
    }

    @Override
    public TShowStorageGroupResp showStorageGroup(List<String> storageGroupPathPattern) {
        return configManager.showStorageGroup(new GetStorageGroupPlan(storageGroupPathPattern));
    }

    @Override
    public TSStatus createSchemaTemplate(TCreateSchemaTemplateReq req) {
        return configManager.createSchemaTemplate(req);
    }

    @Override
    public TGetAllTemplatesResp getAllTemplates() {
        return configManager.getAllTemplates();
    }

    @Override
    public TGetTemplateResp getTemplate(String req) {
        return configManager.getTemplate(req);
    }

    @Override
    public TSStatus setSchemaTemplate(TSetSchemaTemplateReq req) {
        return configManager.setSchemaTemplate(req);
    }

    @Override
    public TGetPathsSetTemplatesResp getPathsSetTemplate(String req) {
        return configManager.getPathsSetTemplate(req);
    }

    @Override
    public TSStatus deactivateSchemaTemplate(TDeactivateSchemaTemplateReq req) {
        return configManager.deactivateSchemaTemplate(req);
    }

    @Override
    public TSStatus unsetSchemaTemplate(TUnsetSchemaTemplateReq req) throws TException {
        return configManager.unsetSchemaTemplate(req);
    }

    @Override
    public TSStatus dropSchemaTemplate(String req) throws TException {
        return configManager.dropSchemaTemplate(req);
    }

    @Override
    public TSStatus deleteTimeSeries(TDeleteTimeSeriesReq req) {
        return configManager.deleteTimeSeries(req);
    }

    @Override
    public TSStatus createPipeSink(TPipeSinkInfo req) {
        return configManager.createPipeSink(new CreatePipeSinkPlan(req));
    }

    @Override
    public TSStatus dropPipeSink(TDropPipeSinkReq req) {
        return configManager.dropPipeSink(new DropPipeSinkPlan(req.getPipeSinkName()));
    }

    @Override
    public TGetPipeSinkResp getPipeSink(TGetPipeSinkReq req) {
        return configManager.getPipeSink(req);
    }

    @Override
    public TSStatus createPipe(TCreatePipeReq req) {
        return configManager.createPipe(req);
    }

    @Override
    public TSStatus startPipe(String pipeName) {
        return configManager.startPipe(pipeName);
    }

    @Override
    public TSStatus stopPipe(String pipeName) {
        return configManager.stopPipe(pipeName);
    }

    @Override
    public TSStatus dropPipe(String pipeName) {
        return configManager.dropPipe(pipeName);
    }

    @Override
    public TShowPipeResp showPipe(TShowPipeReq req) {
        return configManager.showPipe(req);
    }

    @Override
    public TGetAllPipeInfoResp getAllPipeInfo() {
        return configManager.getAllPipeInfo();
    }

    @Override
    public TSStatus recordPipeMessage(TRecordPipeMessageReq req) {
        return configManager.recordPipeMessage(req);
    }

    @Override
    public TGetRegionIdResp getRegionId(TGetRegionIdReq req) {
        if (req.isSetTimeSlotId() && req.getType() != TConsensusGroupType.DataRegion) {
            return new TGetRegionIdResp(new TSStatus(TSStatusCode.ILLEGAL_PARAMETER.getStatusCode()));
        }
        return configManager.getRegionId(req);
    }

    @Override
    public TGetTimeSlotListResp getTimeSlotList(TGetTimeSlotListReq req) {
        long startTime = req.isSetStartTime() ? req.getStartTime() : Long.MIN_VALUE;
        long endTime = req.isSetEndTime() ? req.getEndTime() : Long.MAX_VALUE;
        GetTimeSlotListPlan plan =
                new GetTimeSlotListPlan(req.getStorageGroup(), req.getSeriesSlotId(), startTime, endTime);
        return configManager.getTimeSlotList(plan);
    }

    @Override
    public TGetSeriesSlotListResp getSeriesSlotList(TGetSeriesSlotListReq req) {
        TConsensusGroupType type =
                req.isSetType() ? req.getType() : TConsensusGroupType.ConfigNodeRegion;
        GetSeriesSlotListPlan plan = new GetSeriesSlotListPlan(req.getStorageGroup(), type);
        return configManager.getSeriesSlotList(plan);
    }

    @Override
    public TSStatus migrateRegion(TMigrateRegionReq req) {
        return configManager.migrateRegion(req);
    }

    @Override
    public TSStatus createCQ(TCreateCQReq req) {
        return configManager.createCQ(req);
    }

    @Override
    public TSStatus dropCQ(TDropCQReq req) {
        return configManager.dropCQ(req);
    }

    @Override
    public TShowCQResp showCQ() {
        return configManager.showCQ();
    }

    @Override
    public TSStatus createModel(TCreateModelReq req) throws TException {
        // TODO
        throw new TException(new UnsupportedOperationException().getCause());
    }

    @Override
    public TSStatus dropModel(TDropModelReq req) throws TException {
        // TODO
        throw new TException(new UnsupportedOperationException().getCause());
    }

    @Override
    public TShowModelResp showModel(TShowModelReq req) throws TException {
        // TODO
        throw new TException(new UnsupportedOperationException().getCause());
    }

    @Override
    public TShowTrailResp showTrail(TShowTrailReq req) throws TException {
        // TODO
        throw new TException(new UnsupportedOperationException().getCause());
    }

    @Override
    public TSStatus updateModelInfo(TUpdateModelInfoReq req) throws TException {
        // TODO
        throw new TException(new UnsupportedOperationException().getCause());
    }
}
